package com.hitao.StormCommonPattern;

import com.hitao.bolts.BatcherBolt;
import com.hitao.bolts.IntermediateRankingsBolt;
import com.hitao.bolts.RollingCountBolt;
import com.hitao.bolts.TotalRankingsBolt;
import com.hitao.spouts.WordReader;
import com.hitao.util.StormRunner;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.testing.TestWordSpout;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * Created by Administrator on 2016/5/3.
 */
public class BatchPattern {
    private static final Logger LOG = Logger.getLogger(TopNPattern.class);
    private static final int DEFAULT_RUNTIME_IN_SECONDS = 50;

    private final TopologyBuilder builder;
    private final String topologyName;
    private final Config topologyConfig;
    private final int runtimeInSeconds;

    public BatchPattern(String topologyName) throws InterruptedException {
        builder = new TopologyBuilder();
        this.topologyName = topologyName;
        topologyConfig = createTopologyConfiguration();
        runtimeInSeconds = DEFAULT_RUNTIME_IN_SECONDS;

        wireTopology();
    }

    private static Config createTopologyConfiguration() {
        Config conf = new Config();
        conf.setDebug(false);
        return conf;
    }

    private void wireTopology() throws InterruptedException {
        String spoutId = "wordGenerator";
        String batcherId = "counter";

        builder.setSpout(spoutId, new WordReader(), 5);
        builder.setBolt(batcherId, new BatcherBolt(5, 60), 4).shuffleGrouping(spoutId);
    }

    public void runLocally() throws InterruptedException {
        StormRunner.runTopologyLocally(builder.createTopology(), topologyName, topologyConfig, runtimeInSeconds);
    }

    public void runRemotely() throws Exception {
        StormRunner.runTopologyRemotely(builder.createTopology(), topologyName, topologyConfig);
    }

    public static void main(String[] args) throws Exception {
        String topologyName = "BatchPattern";
        if (args.length >= 1) {
            topologyName = args[0];
        }
        boolean runLocally = true;
        if (args.length >= 2 && args[1].equalsIgnoreCase("remote")) {
            runLocally = false;
        }

        LOG.info("Topology name: " + topologyName);
        BatchPattern rtw = new BatchPattern(topologyName);
        if (runLocally) {
            LOG.info("Running in local mode");
            rtw.runLocally();
        }
        else {
            LOG.info("Running in remote (cluster) mode");
            rtw.runRemotely();
        }
    }
}

